import pandas as pd
from mysql_in import mysql_sql_df
from mysql_in import mysql_to_sql
from pymysql import Connect
def fact_rental():
    conn = Connect(host='localhost', port=3306, user='root', password='123456', database='sakila_dwh_2', charset='utf8')

    # 显示所有列
    pd.set_option('display.max_columns', None)
    # 获取租借表rental
    df_rental=mysql_sql_df('rental')
    #拆分rental_date数据吧日期与时间分开，并且改变格式2020-11-13 16:28:39 变为20201113 162839,并且计算出时间差
    a=[]
    def f(x):
       #因为归还数据有部分人未还，无数据，所以设置判断，如果无数据填充空值
       if len(str(x))>10:
            #把数据按' '空格分开放入a中
            a.append(str(x).split(' '))
            #把刚放入的数据进行替换改变格式
            a[-1][0]=a[-1][0].replace('-','')
            a[-1][1]=a[-1][1].replace(':','')
       else:
           a.append('')
    df_rental['rental_date'].map(f)
    df_time=pd.DataFrame(a)
    # 把代表日期的字段改成日期格式
    df_rental['rental_date']=df_time[0].astype('datetime64')
    df_rental['rental_time_key']=df_time[1]
    df_rental['rental_date_c']=df_time[0]
    a=[]
    df_rental['return_date'].map(f)
    df_time=pd.DataFrame(a)
    # 把代表日期的字段改成日期格式
    df_rental['return_date']=df_time[0].astype('datetime64')
    df_rental['return_time_key']=df_time[1]
    df_rental['rental_duration']=df_rental['return_date']-df_rental['rental_date']
    df_rental['return_date']=df_time[0]
    df_rental['rental_date']=df_rental['rental_date_c']
    df_rental['rental_duration']=df_rental['rental_duration'].astype(str)



    # 获取inventory中的'film_id','store_id'两个字段
    df_inventory=mysql_sql_df('inventory')
    df_inventory=df_inventory[['inventory_id','film_id','store_id']]

    df_rental=pd.merge(left=df_rental,right=df_inventory,how='left',on='inventory_id')
    # print(df_rental)

    # 准备sql
    sql = f"select * from dim_film"
    # 读取数据
    df_fim=pd.read_sql(sql, conn)
    df_fim=df_fim.rename(columns={"index":"film_key"})

    # print(df_fim)
    # df_fim=df_fim[['film_key','film_id','store_id',]]
    #
    df_rental=pd.merge(left=df_rental,right=df_fim,how='left',on='film_id')

    # 查询链接payment表中的支付金额amount
    df_payment=mysql_sql_df('payment')
    df_payment=df_payment[['rental_id','amount']]
    df_rental=pd.merge(left=df_rental,right=df_payment,how='left',on='rental_id')
    # print(df_rental)

    # 查询维度表dim_coustomter
    sql = f"select * from dim_customer"
    dim_customer=pd.read_sql(sql, conn)
    dim_customer=dim_customer.rename(columns={"index":"customer_key"})
    dim_customer=dim_customer[['customer_key','customer_id']]
    # print(dim_customer)
    df_rental=pd.merge(left=df_rental,right=dim_customer,how='left',on='customer_id')

    # 查询维度表dim_staff
    sql = f"select * from dim_staff"
    df_staff=pd.read_sql(sql, conn)
    df_staff=df_staff.rename(columns={"index":"staff_key"})
    df_staff=df_staff[['staff_key','staff_id']]
    df_rental=pd.merge(left=df_rental,right=df_staff,how='left',on='staff_id')
    # print(df_rental)

    # 查询维度表dim_store
    sql = f"select * from dim_store"
    df_store=pd.read_sql(sql, conn)
    df_store=df_store.rename(columns={"index":"store_key"})
    df_store=df_store[['store_key','store_id']]
    df_rental=pd.merge(left=df_rental,right=df_store,how='left',on='store_id')
    # print(df_rental)


    def f(x):
        if len(str(x))>4:
            x=1
        else:
            x=0
        return x
    df_rental['count_returns']=df_rental['return_date'].map(f)

    df_rental['count_rentals']=1
    df_rental['k']=df_rental['return_date']


    df_rental['return_date']=df_rental['k']
    df_rental['rental_duration']=df_rental['rental_duration_x'].map(lambda x:x.split(' ')[0]).map(lambda x: x if x!='NaT' else None )

    df_rental=df_rental[['customer_key','staff_key','film_key','store_key','rental_date','return_date','rental_time_key','count_returns','count_rentals','rental_duration','last_update_y','rental_id','amount']]
    mysql_to_sql(df_rental,'fact_rental')
